package cn.itcast.a_demo;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 实时流处理
 * 接受socket实时数据
 */
public class StreamWordCount {

    public static void main(String[] args) throws Exception {

        /**
         * Flink是一个DAG
         * 1.获取流处理的执行环境
         * 2.加载数据源
         * 3.数据转换
         * 4.数据输出 （print,不是触发算子）
         * 5.触发执行
         */

        /**
         * 自动代码补全： ctrl+alt+v
         * .var 再回车
         */
        //1.获取流处理的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //2.加载数据源
        DataStreamSource<String> source = env.socketTextStream("192.168.88.161", 8090);

        //3.数据转换
        source.map(new MapFunction<String, Tuple2<String ,Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
                return Tuple2.of(value,1);
            }
        }).keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
            @Override
            public String getKey(Tuple2<String, Integer> value) throws Exception {
                return value.f0;
            }
        }) //在流处理中，分组用keyBy
                .sum(1)
                .print();  //4.数据输出 （print,不是触发算子）

        //5.触发执行
        env.execute();
    }
}
